package org.mozhu.simple.rpc.server.network;

import lombok.extern.slf4j.Slf4j;
import org.mozhu.simple.rpc.server.Acceptor;
import org.mozhu.simple.rpc.server.EndPoint;
import org.mozhu.simple.rpc.server.Processor;
import org.mozhu.simple.rpc.server.RpcServerConfig;
import org.mozhu.simple.rpc.utils.ThreadUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class SocketServer {
    private RpcServerConfig config;

    private int numProcessorThreads;
    private int maxQueuedRequests;

    private RequestChannel requestChannel;
    private List<Processor> processors;
    private Map<EndPoint, Acceptor> acceptors;

    private int sendBufferSize;
    private int recvBufferSize;

    public SocketServer(RpcServerConfig config) {
        this.config = config;
        numProcessorThreads = config.getNumNetworkThreads();
        maxQueuedRequests = config.getQueuedMaxRequests();
        sendBufferSize = config.getSocketSendBufferBytes();
        recvBufferSize = config.getSocketReceiveBufferBytes();
    }

    public synchronized void startup() {
        requestChannel = new RequestChannel(numProcessorThreads, maxQueuedRequests);
        processors = new ArrayList<>(numProcessorThreads);
        acceptors = new HashMap<>();

        for (int i = 0; i < numProcessorThreads; i++) {
            Processor processor = new Processor(i, requestChannel);
            processors.add(processor);
        }

        EndPoint endPoint = new EndPoint(config.getHost(), config.getPort());
        Acceptor acceptor = new Acceptor(endPoint, sendBufferSize, recvBufferSize, processors);
        acceptors.put(endPoint, acceptor);

        ThreadUtils.newThread(String.format("rpc-socket-acceptor-", endPoint.getHost(), endPoint.getPort()), acceptor, false).start();
        acceptor.awaitStartup();

        log.info("Started " + acceptors.size() + " acceptor threads");
    }
}
